package application

import (
	"fmt"
	log "github.com/sirupsen/logrus"
	"go.uber.org/atomic"
	"net"
	"sync"
	"time"
	"usbipproxy/common"
	"usbipproxy/domain"
	"usbipproxy/domain/service"
	"usbipproxy/domain/tcpsession"
	"usbipproxy/pb"
)

type TcpServerServiceImpl struct {
	listenAddr       string
	listener         net.Listener
	rwLock           sync.RWMutex
	requests         map[string]chan *pb.ResponseParam
	msgNotifyService service.MsgNotifyService
	closed           atomic.Bool
	quit             chan struct{}
	tcpMsgHandle     TcpMsgHandleService
	usbIpType        pb.UsbIpType
	sessMgr          tcpsession.TcpSessionMgrRepo
	swService        service.SwDeviceService
}

func (t *TcpServerServiceImpl) Start() error {
	//addr := fmt.Sprintf("%s:%d", t.host, t.port)
	listener, err := net.Listen("tcp", t.listenAddr)
	if err != nil {
		log.Errorf("TcpServerServiceImpl start tcp listener addr:%s error: %s", t.listenAddr, err.Error())
		return err
	}

	t.listener = listener
	//defer listener.Close()
	log.Infof("Server listening on %s", t.listenAddr)

	go func() {
		for {
			// 接受客户端连接
			conn, err := listener.Accept()
			if err != nil {
				log.Infof("Error accepting connection:", err)
				continue
			}
			tcpConn, ok := conn.(*net.TCPConn)
			if !ok {
				log.Errorf("Connection is not a TCP connection")
			}

			// 设置TCP_NODELAY选项
			//if err := tcpConn.SetNoDelay(true); err != nil {
			//	log.Errorf("Failed to set TCP_NODELAY: %v", err)
			//}
			if err := tcpConn.SetReadBuffer(1024 * 1024); err != nil {
				log.Errorf("Failed to set SetReadBuffer: %v", err)
			}
			if err := tcpConn.SetWriteBuffer(1024 * 1024); err != nil {
				log.Errorf("Failed to set SetWriteBuffer: %v", err)
			}

			go func() {
				log.Infof("a new conn:%s come", conn.RemoteAddr().String())
				connId, _ := common.Snowflake.GetSnowflakeId()
				if t.usbIpType == pb.UsbIpType_USBIP_TYPE_CLIENT && t.swService != nil {
					t.swService.StartUsbBusControl(connId, 0)
				}
				session := tcpsession.NewTcpSession(connId, conn, domain.NewMsgPipe())
				dCli := domain.NewDispatchClient("", t.usbIpType, session)
				t.sessMgr.AddConn(connId, dCli)
				go session.Start()
				t.handleDispatchCliMsg(dCli)
				if t.usbIpType == pb.UsbIpType_USBIP_TYPE_CLIENT && t.swService != nil {
					t.swService.StopUsbBusControl(connId)
				}
			}()

		}
	}()

	go t.handleNotifyMsg()

	return nil
}

func (t *TcpServerServiceImpl) Stop() error {
	if !t.closed.CompareAndSwap(false, true) {
		return nil
	}
	//tcp session close
	t.listener.Close()
	close(t.quit)
	return nil
}

func (t *TcpServerServiceImpl) handleDispatchCliMsg(dCli *domain.DispatchClient) {
LOOP:
	for {
		select {
		case <-t.quit:
			break LOOP
		case msg := <-dCli.GetSessRepo().GetInMsg():
			//log.Debugf("handRecvAndSend msg_type:%v", msg.GetMsgType())
			if msg == nil {
				log.Infof("TcpServerServiceImpl handRecvAndSend msg is nil")
				t.sessMgr.DelByConnId(dCli.GetConnId())
				//给自己发一个 attach over 的消息
				t.SendAttachOverMsgSelf(dCli)
				break LOOP
			}
			dCli.UpdateBaseParamWithToId(msg, dCli.GetConnId())
			switch msg.GetMsgType() {
			case pb.MsgType_RESPONSE:
				t.HandleResponseMsg(msg, dCli)
			case pb.MsgType_LOGIN:
				t.HandleLoginMsg(msg, dCli)
				//t.tcpMsgHandle.Handle(msg, dCli)
			default:
				t.tcpMsgHandle.Handle(msg, dCli)
			}
		}
	}

	log.Infof("TcpServerServiceImpl handleDispatchCliMsg handRecvAndSend handle msg over")
}

func (t *TcpServerServiceImpl) handleNotifyMsg() {
LOOP:
	for {
		select {
		case <-t.quit:
			break LOOP
		case msg := <-t.msgNotifyService.GetOutMsgChan():
			toId := msg.GetBase().GetToId()
			dCli := t.sessMgr.GetByConnId(toId)
			if dCli == nil {
				log.Infof("TcpServerServiceImpl GetOutMsgChan handleNotifyMsg connId:%v can not find dispatch client", toId)
				continue
			}
			dCli.UpdateBaseParamForOut(msg)
			dCli.GetSessRepo().SendOutMsg(msg)
		case msg := <-t.msgNotifyService.GetInMsgChan():
			toId := msg.GetBase().GetToId()
			dCli := t.sessMgr.GetByConnId(toId)
			if dCli == nil {
				log.Infof("TcpServerServiceImpl GetInMsgChan handleNotifyMsg connId:%v can not find dispatch client", toId)
				continue
			}
			dCli.GetSessRepo().RecvInMsg(msg)
		}
	}

	log.Infof("TcpServerServiceImpl handleNotifyMsg handRecvAndSend handle msg over")
}

func (t *TcpServerServiceImpl) SendAttachOverMsgSelf(dCli *domain.DispatchClient) {
	var msg *pb.ProxyMsg
	if t.usbIpType == pb.UsbIpType_USBIP_TYPE_SERVER {
		msg = &pb.ProxyMsg{
			MsgType: pb.MsgType_TO_SERVER_ATTACH_OVER,
			Base:    &pb.BaseParam{ToId: dCli.GetConnId()},
			Param: &pb.ProxyMsg_ToServerAttachOverParam{ToServerAttachOverParam: &pb.ToServerAttachOverParam{
				BusId: "all",
			}},
		}
	} else {
		msg = &pb.ProxyMsg{
			MsgType: pb.MsgType_TO_CLIENT_ATTACH_OVER,
			Base:    &pb.BaseParam{ToId: dCli.GetConnId()},
			Param:   &pb.ProxyMsg_ToClientAttachOverParam{ToClientAttachOverParam: &pb.ToClientAttachOverParam{}},
		}
	}

	t.tcpMsgHandle.Handle(msg, dCli)
}

func (t *TcpServerServiceImpl) SendOutMsgWithNoReply(msg *pb.ProxyMsg) error {
	return t.msgNotifyService.SendMsg(msg)
}

func (t *TcpServerServiceImpl) SendOutMsg(msg *pb.ProxyMsg) (*pb.ResponseParam, error) {
	t.msgNotifyService.SendMsg(msg)
	respChan := make(chan *pb.ResponseParam, 1)
	t.rwLock.Lock()
	t.requests[msg.GetBase().GetReqId()] = respChan
	t.rwLock.Unlock()

	// 等待响应或超时
	select {
	case resp := <-respChan:
		log.Infof("TcpClientServiceImpl SendMsg recv rsp %v", msg.GetBase().GetReqId())
		return resp, nil
	case <-time.After(time.Duration(common.ResponseTimeout) * time.Second):
		t.removeRequest(msg.GetBase().GetReqId())
		log.Infof("TcpClientServiceImpl req:%v SendMsg timeout", msg.GetBase().GetReqId())
		return nil, fmt.Errorf("request:%v timeout", msg.GetBase().GetReqId())
	}
}

func (t *TcpServerServiceImpl) SendOutMsgByDispatchId(msg *pb.ProxyMsg, dispatchId string) (*pb.ResponseParam, error) {
	cli := t.sessMgr.GetByDid(dispatchId)
	if cli == nil {
		log.Warnf("TcpServerServiceImpl SendOutMsgByDispatchId dispatch client:%v not find dispatch", dispatchId)
		return nil, fmt.Errorf("dispatch client:%v not find dispatch", dispatchId)
	}

	log.Infof("SendOutMsgByDispatchId find connId:%v by dispatchId:%v", cli.GetConnId(), dispatchId)
	cli.UpdateBaseParamWithToId(msg, cli.GetConnId())
	return t.SendOutMsg(msg)
}

func (t *TcpServerServiceImpl) SendOutMsgByDispatchIdWithNoReply(msg *pb.ProxyMsg, dispatchId string) error {
	cli := t.sessMgr.GetByDid(dispatchId)
	if cli == nil {
		log.Warnf("TcpServerServiceImpl SendOutMsgByDispatchId dispatchId:%v not find dispatch", dispatchId)
		return fmt.Errorf("dispatchId:%v not find dispatch", dispatchId)
	}

	log.Infof("SendOutMsgByDispatchIdWithNoReply find connId:%v by dispatchId:%v", cli.GetConnId(), dispatchId)
	cli.UpdateBaseParamWithToId(msg, cli.GetConnId())
	return t.SendOutMsgWithNoReply(msg)
}

func (t *TcpServerServiceImpl) removeRequest(requestID string) {
	t.rwLock.Lock()
	delete(t.requests, requestID)
	t.rwLock.Unlock()
}

func (t *TcpServerServiceImpl) RecvInMsg(msg *pb.ProxyMsg) error {
	t.msgNotifyService.RecvMsg(msg)
	return nil
}

func (t *TcpServerServiceImpl) RecvInMsgByDispatchId(msg *pb.ProxyMsg, dispatchId string) error {
	cli := t.sessMgr.GetByDid(dispatchId)
	if cli == nil {
		log.Warnf("TcpServerServiceImpl RecvInMsgByDispatchId dispatch client:%v not find dispatch", dispatchId)
		return fmt.Errorf("dispatch client:%v not find dispatch", dispatchId)
	}

	log.Infof("TcpServerServiceImpl RecvInMsgByDispatchId find connId:%v by dispatchId:%v", cli.GetConnId(), dispatchId)
	cli.UpdateBaseParamWithToId(msg, cli.GetConnId())
	return t.RecvInMsg(msg)
}

func (t *TcpServerServiceImpl) HandleResponseMsg(msg *pb.ProxyMsg, dCli *domain.DispatchClient) error {
	reqId := msg.GetBase().GetReqId()
	t.rwLock.RLock()
	rspChan := t.requests[reqId]
	t.rwLock.RUnlock()

	if rspChan != nil {
		select {
		case rspChan <- msg.GetResponseParam():
			log.Debugf("recv response reqId:%v", reqId)
		default:
			log.Warnf("reqId:%v response channel has been sended or closed", reqId)
		}
	}

	return nil
}

func (t *TcpServerServiceImpl) HandleLoginMsg(msg *pb.ProxyMsg, dCli *domain.DispatchClient) error {

	did := msg.GetLoginParam().GetDid()
	log.Infof("TcpServerServiceImpl HandleLoginMsg dispatch id:%v", did)
	if did == "" {
		return nil
	}
	dCli.SetDid(did)
	t.sessMgr.AddDispatch(did, dCli)

	return nil
}

func NewTcpServerServiceImpl(listenAddr string, usbIpType pb.UsbIpType,
	msgNotifyService service.MsgNotifyService,
	tcpMsgHandle TcpMsgHandleService,
	sessMgr tcpsession.TcpSessionMgrRepo, swService service.SwDeviceService) TcpServerService {
	return &TcpServerServiceImpl{
		listenAddr:       listenAddr,
		requests:         make(map[string]chan *pb.ResponseParam),
		msgNotifyService: msgNotifyService,
		quit:             make(chan struct{}, 1),
		tcpMsgHandle:     tcpMsgHandle,
		usbIpType:        usbIpType,
		sessMgr:          sessMgr,
		swService:        swService,
	}
}
